package ex.datastream.watermark;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import javax.annotation.Nullable;

public class TimeTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        //设置为eventime事件类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置水印生成时间间隔100ms
        env.getConfig().setAutoWatermarkInterval(100);
        DataStream<String> dataStream = env.socketTextStream("localhost", 9999, "\n")
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
                    private Long currentTimeStamp = 0L;
                    //设置允许乱序时间5秒
                    private Long maxoutOfOrderness = 0L;

                    @Nullable
                    @Override
                    public Watermark getCurrentWatermark() {
                        return new Watermark(currentTimeStamp);
                    }

                    @Override
                    public long extractTimestamp(String s, long l) {
                        String[] arr = s.split(",");
                        long timeStamp = Long.parseLong(arr[1]);
                        currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
                        System.err.println(s + "EventTime:" + timeStamp + ",watermark:" + (currentTimeStamp - maxoutOfOrderness));
                        return timeStamp;
                    }
                });
                dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String s) throws Exception {
                        String[] arr = s.split(",");
                        return new Tuple2<String, Long>(arr[0], Long.parseLong(arr[1]));
                    }
                })
                .keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .minBy(1);
        dataStream.print();
        env.execute("WaterMark Test Demo");
    }
//第一条数据的时间戳为 1588659181000
//窗口的大小为5秒
//那么应该会在 flink,1588659185000
//这条数据出现时触发窗口的计算

//flink,1588659181000
//flink,1588659182000
//flink,1588659183000
//flink,1588659184000
//flink,1588659185000
//
//
//    flink,1588659181000
//    flink,1588659182000
//    flink,1588659183000
//    flink,1588659184000
//    flink,1588659185000
//    flink,1588659180000 乱序
//    flink,1588659186000
//    flink,1588659187000
//    flink,1588659188000
//    flink,1588659189000
//    flink,1588659190000
//Flink 在用时间+窗口+水印来解决实际生产中的数据乱序问题，有如下的触发条件:
//    第一步:
//    watermark时间>=window_end time
//    第二步:
//    在 [window_start_time,window_end_time)中有数据存在，这个窗口是左闭右开的

}
